<?php

namespace Zeke\Io\Stream;

use Zeke\Zee\Event\BufferEvent;
use Zeke\Zee\Event\EventEmitter;
use Zeke\Contracts\Io\Loop\LoopContract;
use Zeke\Contracts\Io\Stream\WritableStreamContracts;
use Zeke\Zee\Event\IoEvent;


final class WritableResourceStream extends EventEmitter implements WritableStreamContracts
{


    /**
     * @var resource
     */
    private $stream;

    /**
     * @var LoopContract
     */
    private $loop;


    /**
     * writeStream设置的数据缓冲区
     * @var int
     */
    private $softLimit;


    /**
     * PHP_VERSION_ID < 70018 || (PHP_VERSION_ID >= 70100 && PHP_VERSION_ID < 70104)
     * fwrite 一次最大允许发送8192字节，超过需要分段发送
     * @var int
     */
    private $writeChunkSize;


    /**
     * @var bool
     */
    private $listening = false;


    /**
     * 默认为true,如果应用程序主动调用end()或close()方法，则将此值设置为false
     * 当调用write()后，如果为此值false，则写入失败，在write回调用，则写完后自动调用close()，关闭流
     * @var bool
     */
    private $writable = true;


    /**
     * 默认为false,此值在调用close()后设置为true,主要目的防止父类关闭事件触发多次调用关闭流
     * @var bool
     */
    private $closed = false;


    /**
     * @var string
     */
    private $data = '';


    /**
     * WritableResourceStream constructor.
     * @param $stream
     * @param LoopContract $loop
     * @param null $writeBufferSoftLimit
     * @param null $writeChunkSize
     */
    public function __construct($stream, LoopContract $loop, $writeBufferSoftLimit = null, $writeChunkSize = null)
    {
        if (!is_resource($stream) || get_resource_type($stream) !== "stream") {
            throw new \InvalidArgumentException('First parameter must be a valid stream resource');
        }

        // ensure resource is opened for writing (fopen mode must contain either of "waxc+")
        $meta = stream_get_meta_data($stream);
        if (isset($meta['mode']) && $meta['mode'] !== '' && strtr($meta['mode'], 'waxc+', '.....') === $meta['mode']) {
            throw new \InvalidArgumentException('Given stream resource is not opened in write mode');
        }

        // this class relies on non-blocking I/O in order to not interrupt the event loop
        // e.g. pipes on Windows do not support this: https://bugs.php.net/bug.php?id=47918
        if (stream_set_blocking($stream, 0) !== true) {
            throw new \RuntimeException('Unable to set stream resource to non-blocking mode');
        }

        $this->stream           = $stream;
        $this->loop             = $loop;
        $this->softLimit        = ($writeBufferSoftLimit === null) ? 65536 : (int)$writeBufferSoftLimit;
        $this->writeChunkSize   = ($writeChunkSize === null) ? -1 : (int)$writeChunkSize;
    }


    /**
     * stream是否可写
     *
     * @return bool
     *
     *
     * @author <liu.bin@ruaby.com> 2018-08-06 22:43
     */
    public function isWritable()
    {
        return $this->writable;
    }


    /**
     * 向stream写入数据，如果可写，会调用系统函数 fwrite
     *
     * @param mixed|string $data
     * @return bool
     *
     * @author <liu.bin@ruaby.com> 2018-08-06 22:43
     */
    public function write($data)
    {
        if (!$this->writable) {
            return false;
        }
        $this->data .= $data;
        if (!$this->listening && $this->data !== '') {
            $this->listening = true;
            $this->loop->addWriteStream($this->stream, array($this, 'handleWrite'));
        }

        //判断字符串长度是否超出buffer限定大小，
        //超过：false
        //不超过：true
        return !isset($this->data[$this->softLimit - 1]);
    }


    /**
     * 关闭
     *
     * @param null $data
     *
     *
     * @author <liu.bin@ruaby.com> 2018-08-06 22:45
     */
    public function end($data = null)
    {
        if (null !== $data) {
            $this->write($data);
        }

        $this->writable = false;

        // close immediately if buffer is already empty
        // otherwise wait for buffer to flush first
        if ($this->data === '') {
            $this->close();
        }
    }


    /**
     * 关闭
     *
     * @author <liu.bin@ruaby.com> 2018-08-06 22:45
     */
    public function close()
    {
        if ($this->closed) {
            return;
        }

        if ($this->listening) {
            $this->listening = false;
            $this->loop->removeWriteStream($this->stream);
        }

        $this->closed = true;
        $this->writable = false;
        $this->data = '';

        $this->emit(IoEvent::CLOSE);
        $this->removeAllListeners();

        if (is_resource($this->stream)) {
            fclose($this->stream);
        }
    }


    /**
     * stream可写后调用此函数
     *
     * @author <liu.bin@ruaby.com> 2018-08-06 22:46
     */
    public function handleWrite()
    {
        $error = null;
        set_error_handler(function ($errNo, $errStr, $errFile, $errLine) use (&$error) {
            $error = array('message' => $errStr, 'number' => $errNo, 'file' => $errFile, 'line' => $errLine);
        });



        if ($this->writeChunkSize === -1) {
            $sent = fwrite($this->stream, $this->data);
        } else {
            //分段发送，发完全部或发送完writeChunkSize数据，fwrite 返回
            $sent = fwrite($this->stream, $this->data, $this->writeChunkSize);
        }

        restore_error_handler();

        // Only report errors if *nothing* could be sent.
        // Any hard (permanent) error will fail to send any data at all.
        // Sending excessive amounts of data will only flush *some* data and then
        // report a temporary error (EAGAIN) which we do not raise here in order
        // to keep the stream open for further tries to write.
        // Should this turn out to be a permanent error later, it will eventually
        // send *nothing* and we can detect this.
        if ($sent === 0 || $sent === false) {
            if ($error !== null) {
                $error = new \ErrorException($error['message'], 0, $error['number'], $error['file'], $error['line']);
            }
            $this->emit('error', array(new \RuntimeException('Unable to write to stream: ' . ($error !== null ? $error->getMessage() : 'Unknown error'), 0, $error)));
            $this->close();
            return;
        }


        /**
         * softLimit        = 65535
         * writeChunkSize   = 8192
         * $exceeded     data是否超出最大发送长度
         * $sent         已发送数据长度
         */
        $exceeded = isset($this->data[$this->softLimit - 1]);


        //数据分割，等待下一个IoEvent::WRITE事件
        $this->data = (string) substr($this->data, $sent);

        // buffer has been above limit and is now below limit
        /**
         * 当前需要发送的总数据 大于 softLimit && 发送后剩余数据 小于 softLimit，触发BufferEvent::DRAIN事件
         *
         * BufferEvent::DRAIN事件:
         * 1:当缓冲区写满数据，并且发送部分数据后，缓冲区有空的容量，则触发BufferEvent::DRAIN事件。
         * 2:缓冲区填满之前，这个事件应该发出一次，现在准备接受更多的数据。
         * 3:换句话说，这个事件可以发生任意次数，如果缓冲区从来没有满过的话，这可能是零次。
         *   如果缓冲区已开始就没满，则不应发出此事件。
         *
         */
        if ($exceeded && !isset($this->data[$this->softLimit - 1])) {
            $this->emit(BufferEvent::DRAIN);
        }


        // buffer is now completely empty => stop trying to write
        if ($this->data === '') {
            // stop waiting for resource to be writable
            if ($this->listening) {
                $this->loop->removeWriteStream($this->stream);
                $this->listening = false;
            }

            // buffer is end()ing and now completely empty => close buffer
            if (!$this->writable) {
                $this->close();
            }
        }
    }
}
